package com.cyf.common.core.thread.executor;

import java.util.List;
import java.util.concurrent.*;

/**
 * 可监控的线程池
 * 考虑不使用包装类去实现，需要重写的方法太多，并且配置线程池也就是这几个参数，构造参数传入即可
 * @author chenyifan
 * @create 2024-08-24 10:33
 */
public class ExecutorMonitor extends ThreadPoolExecutor {

    /**
     * 保存任务开始执行的时间
     */
    private ConcurrentHashMap<Integer, Long> startTimes;

    /**
     * 线程池名称
     */
    private String poolName;

    private static final RejectedExecutionHandler defaultHandler = new CallerRunsPolicy();

    public ExecutorMonitor(String poolName, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        this(poolName, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, defaultHandler);
    }

    public ExecutorMonitor(String poolName, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        this(poolName, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler);
    }

    public ExecutorMonitor(String poolName, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        this(poolName, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler);
    }

    public ExecutorMonitor(String poolName, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        this.startTimes = new ConcurrentHashMap<>();
        this.poolName = poolName;
    }


    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        // 记录任务开始执行时间
        startTimes.put(r.hashCode(), System.currentTimeMillis());
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        //这个是 execute 提交的时候，execute 并不会吞异常
        if (t != null) {
            printMsg("afterExecute 里面获取到 execute 提交的异常信息：" + t.getMessage());
        }

        // 如果 r 的实际类型是 FutureTask，那么是 submit 提交的，如果线程执行出错，异常不会抛出
        // 因为 submit 的 task.run 里面把异常吞了，只有在调 future.get() 才抛出异常，这里提前调用
        if (r instanceof FutureTask) {
            try {
                Future<?> future = (Future<?>) r;
                //get获取异常
                future.get();
            } catch (Exception e) {
                printMsg("afterExecute 里面获取到 submit 提交的异常信息：" + e.getMessage());
            }
        }

        long startTime = startTimes.remove(r.hashCode());
        long endTime = System.currentTimeMillis();
        // 任务耗时
        printMsg(String.format("pool-monitor: 任务耗时: %d ms", endTime -  startTime));
    }

    @Override
    public void shutdown() {
        printMsg(" Going to shutdown");
        super.shutdown();
    }

    @Override
    public List<Runnable> shutdownNow() {
        printMsg(" Going to Stop");
        return super.shutdownNow();
    }

    private void printMsg(String msg) {
        msg += String.format("线程池[%s] ", poolName) + msg;
        System.out.println(msg);
    }
}
